package com.shujia.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo1Source {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
      * 基于本地集合构建DataStream  -- 有界流
      *
      */

    val lisrDS: DataStream[Int] = env.fromCollection(List(1, 2, 3, 4, 5, 6, 7, 8, 9))

    //lisrDS.print()


    /**
      * 基于文件构建DataStream -- 有界流
      *
      */

    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")

    studentDS
      .map(stu => (stu.split(",")(4), 1))
      .keyBy(_._1)
      .sum(1)
    //.print()


    /**
      * 基于socket构建DataStream-- 无界流
      *
      */

    //val socketDS: DataStream[String] = env.socketTextStream("master", 8888)
    //socketDS.print()


    /**
      * 自定义source , 实现SOurceFunction接口
      *
      */

    val myDS: DataStream[Int] = env.addSource(new MySource)

    myDS.print()

    env.execute()


  }

}


/**
  * 自定义source, 实现SourceFunction接口
  * 实现run方法，
  *
  */

class MySource extends SourceFunction[Int] {
  /**
    * run方法只执行一次
    *
    * @param ctx ： 用于发送数据到下游task
    */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    var i = 0
    while (true) {

      //将数据发送到下游
      ctx.collect(i)

      Thread.sleep(50)

      i += 1
    }


  }


  /**
    * 再任务取消的时候执行，用于回收资源
    *
    */
  override def cancel(): Unit = {}
}

